package org.example.flink.window;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUnit;
import java.time.Duration;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.source.S3;
import scala.Tuple2;
import scala.Tuple3;

/**
 * @author zhangjungang
 * @desc FlinkWindow测试
 * @date 2021/10/12 上午10:56
 */
public class WindowFlink {

    private static final String pattern = "yyyy-MM-dd HH:mm:ss";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<Tuple3<String, Date, Integer>> datasource =
                env.addSource(new S3())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, Date, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Date, Integer>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, Date, Integer> tuple3, long l) {
                                return tuple3._2().getTime();
                            }
                        }));


        datasource.keyBy(new KeySelector<Tuple3<String, Date, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, Date, Integer> tuple3) throws Exception {
                return tuple3._1();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .allowedLateness(Time.seconds(1))
                .process(new ProcessWindowFunction<Tuple3<String, Date, Integer>, Object, String, TimeWindow>() {

            int cnt = 0;
            Map<String, Tuple2<Date,Date>> map = new HashMap<>();

            @Override
            public void process(String s, Context context, Iterable<Tuple3<String, Date, Integer>> iterable,
                                Collector<Object> collector) throws Exception {
                Iterator<Tuple3<String, Date, Integer>> iterator = iterable.iterator();

                while (iterator.hasNext()){
                    Tuple3<String, Date, Integer> next = iterator.next();
                    cnt++;
                    System.out.println(next+" "+cnt);

                    Tuple2<Date,Date> tuple2 = map.get(next._1());
                    if(tuple2 == null){
                        map.put(next._1(),Tuple2.apply(next._2(),null));
                    }else {
                        map.remove(next._1());
                        System.out.println(StringUtils.substringAfterLast(Thread.currentThread().getName()," ")+" "+DateTime.of(new Date(context.window().maxTimestamp())).toString(pattern)+"======== "+next._1()+" st="+DateTime.of(tuple2._1()).toString(pattern)+" et="+DateTime.of(tuple2._2()).toString(pattern));
                    }


                }
            }
        }).setParallelism(2);

        env.execute("WindowFlink");

    }

}
